package com.learn.state;


import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @create: 2023-04-20 17:22
 * @author: Mr.Du
 * --------------
 * @notes: 自定义实现State实现单词统计
 * state就是状态，Flink中所有算子都是有状态的计算
 * state根据不同的算子状态，可以划分出来两类state：
 * KeyedState：
 *  应用在keyBy算子之上的State
 *  ValueState->ValueStateDescriptor创建而来
 *  ListState->ListStateDescriptor创建而来
 *  MapState->MapStateDescriptor创建而来
 *  ReducingState->ReducingStateDescriptor创建而来
 *
 * OperatorState:
 *  应用到非keyBy算子之上的State，与并行度的数量有关，有多少个并行度就有多少个OperatorState
 *   ListState->ListStateDescriptor创建而来
 **/
public class KeyedStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //启动checkpoint
        //env.enableCheckpointing(5000L);

        DataStream<String> socketTextStream = env.socketTextStream("node1",9999);
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneStream = socketTextStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });

/*        wordAndOneStream
                .keyBy(0).reduce(new RichReduceFunction<Tuple2<String, Integer>>() {

                    //定义state数据结构
                    private ValueState<Tuple2<String,Integer>> valueState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        //初始化ValueState对象
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Tuple2<String, Integer>>(
                                "reduceState",
                                TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {
                                })
                        ));
                    }

                    *//**
                     *
                     * @param tuple1 初始值或者临时累加值
                     * @param tuple2 下一个元素
                     * @return
                     * @throws Exception
                     *//*
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> tuple1, Tuple2<String, Integer> tuple2) throws Exception {
                        Tuple2<String, Integer> result = valueState.value();
                        if(result == null) {
                            result = Tuple2.of(tuple1.f0,tuple1.f1);
                        }

                        Tuple2<String, Integer> resultSum = Tuple2.of(result.f0, result.f1 + tuple2.f1);
                        valueState.update(resultSum);
                        return resultSum;
                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                        Tuple2<String, Integer> wordSum = valueState.value();
                        System.out.println("===========================");
                        System.out.println(wordSum);
                    }
                }).print();*/

        wordAndOneStream
                .keyBy(t->t.f0)
                .reduce(new RichReduceFunction<Tuple2<String, Integer>>() {
                    //定义valueState
                    private ValueState<Integer> valueState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        valueState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>(
                                "valueState",
                                TypeInformation.of(new TypeHint<Integer>(){})
                        ));
                    }

                    /**
                     *
                     * @param tuple1 初始值
                     * @param tuple2 下一个值
                     * @return
                     * @throws Exception
                     */
                    @Override
                    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> tuple1, Tuple2<String, Integer> tuple2) throws Exception {
                        Integer result = valueState.value();
                        if(result == null) {
                           result = tuple1.f1;
                        }
                        Tuple2<String, Integer> resultSum = Tuple2.of(tuple2.f0,result + tuple2.f1);
                        valueState.update(result + tuple2.f1);
                        return resultSum;
                    }

                    @Override
                    public void close() throws Exception {

                    }
                }).print();
        env.execute();
    }
}
